package day08;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Description: 流 -> 动态表
 * @Author ZYX
 * @Date 2022/2/20 20:06
 * @Version 1.0
 */
public class Demo07 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 数据源，且设置水位线
        SingleOutputStreamOperator<Tuple3<String, String, Long>> stream = env.fromElements(
                Tuple3.of("Mary", "./home", 12 * 60 * 60 * 1000L),
                Tuple3.of("Bob", "./cart", 12 * 60 * 60 * 1000L),
                Tuple3.of("Mary", "./prod?id=1", 12 * 60 * 60 * 1000L + 5 * 1000L),
                Tuple3.of("Liz", "./home", 12 * 60 * 60 * 1000L + 60 * 1000L),
                Tuple3.of("Bob", "./prod?id=3", 12 * 60 * 60 * 1000L + 90 * 1000L),
                Tuple3.of("Mary", "./prod?id=7", 12 * 60 * 60 * 1000L + 105 * 1000L)
        ).assignTimestampsAndWatermarks(
                WatermarkStrategy.<Tuple3<String, String, Long>>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<Tuple3<String, String, Long>>() {
                            @Override
                            public long extractTimestamp(Tuple3<String, String, Long> stringStringLongTuple3, long l) {
                                return stringStringLongTuple3.f2;
                            }
                        })
        );

        // 创建表环境
        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, environmentSettings);

        // 数据流 -> 动态表
        Table table = tableEnv.fromDataStream(
                stream,
                $("f0").as("user"),
                $("f1").as("url"),
                // 使用rowtime方法指定f2字段为事件时间
                $("f2").rowtime().as("cTime")

        );

        // 动态表 -> 数据流
        tableEnv.toDataStream(table).print();

        env.execute();
    }

}
